package com.tianji.remark.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tianji.api.dto.remark.LikedTimesDTO;
import com.tianji.common.autoconfigure.mq.RabbitMqHelper;
import com.tianji.common.utils.StringUtils;
import com.tianji.common.utils.UserContext;
import com.tianji.remark.constants.RedisConstants;
import com.tianji.remark.domain.dto.LikeRecordFormDTO;
import com.tianji.remark.domain.po.LikedRecord;
import com.tianji.remark.mapper.LikedRecordMapper;
import com.tianji.remark.service.ILikedRecordService;
import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static com.tianji.common.constants.MqConstants.Exchange.LIKE_RECORD_EXCHANGE;
import static com.tianji.common.constants.MqConstants.Key.LIKED_TIMES_KEY_TEMPLATE;
import static com.tianji.remark.constants.RedisConstants.LIKES_TIMES_KEY_PREFIX;
import static com.tianji.remark.constants.RedisConstants.LIKE_BIZ_KEY_PREFIX;

/**
 * <p>
 * 点赞记录表 服务实现类
 * </p>
 *
 * @author xioayang
 * @since 2024-06-28
 */
@Service
@RequiredArgsConstructor
public class LikedRecordServiceRedisImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {

    private final RabbitMqHelper mqHelper;
    private final StringRedisTemplate redisTemplate;
    /**
     * 点赞或取消点赞
     *
     * @param recordDTO 点赞业务类型
     */
    @Override
    public void addLikeRecord(LikeRecordFormDTO recordDTO) {
        // 判断是否点赞,
        Boolean isliked = recordDTO.getLiked() ? like(recordDTO) : unLike(recordDTO);
        //判断执行是否成功
        if (!isliked){
            return;
        }
        //统计点赞数
        Long likedTimes = redisTemplate.opsForSet().size(LIKE_BIZ_KEY_PREFIX + recordDTO.getBizId());

        //统计点赞总数，储存到redis的Zset集合中
        redisTemplate.opsForZSet().add(LIKES_TIMES_KEY_PREFIX+recordDTO.getBizType(), recordDTO.getBizId().toString(), likedTimes);

    }

    /**
     * 批量查询点赞状态
     *
     * @param bizIds 点赞业务id
     * @return 点赞状态
     */
    @Override
    public Set<Long> isBizLiked(List<Long> bizIds) {
        //获取用户id
        Long userId = UserContext.getUser();
        //使用sismember命令查询判断是否已点赞,executePipelined可以批量处理
        List<Object> pipelinedRedisResult = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection connect = (StringRedisConnection) connection;
            for (Long bizId : bizIds) {
                connect.sIsMember(LIKE_BIZ_KEY_PREFIX + bizId, userId.toString());
            }
            return null;
        });
        //stream流收集已点赞的业务id并返回
        return IntStream.range(0,pipelinedRedisResult.size())
                .filter(i -> (Boolean) pipelinedRedisResult.get(i))
                .mapToObj(i -> bizIds.get(i))
                .collect(Collectors.toSet());
    }

    /**
     * 读取点赞次数并发送消息
     *
     * @param bizType    业务类型
     * @param maxBizSize 最大业务数量
     */
    @Override
    public void readLikeTimesAndSendMessage(String bizType, Integer maxBizSize) {
        //获取点赞数量
        Set<ZSetOperations.TypedTuple<String>> typedTuples = redisTemplate.opsForZSet().popMin(LIKES_TIMES_KEY_PREFIX + bizType, maxBizSize);

       while (ObjectUtil.isNotEmpty(typedTuples)) {
           //封装数据
           List<LikedTimesDTO> messages = new ArrayList<>(typedTuples.size());
           for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
               String bizId = typedTuple.getValue();
               Double liketimes = typedTuple.getScore();
               messages.add(LikedTimesDTO.of(Long.valueOf(bizId), liketimes.intValue()));
           }
           //发送MQ消息
           mqHelper.send(LIKE_RECORD_EXCHANGE,
                   StringUtils.format(LIKED_TIMES_KEY_TEMPLATE + bizType),
                   messages);
           //查询下一轮消息（30条数据）
           typedTuples = redisTemplate.opsForZSet().popMin(LIKES_TIMES_KEY_PREFIX + bizType, maxBizSize);

       }

    }

    /**
     * 取消点赞
     * @param recordDTO 点赞业务类型
     * @return 是否取消点赞
     */
    private Boolean unLike(LikeRecordFormDTO recordDTO) {
        //1.获取用户id
        Long userId = UserContext.getUser();
        //2.获取key
        String key = LIKE_BIZ_KEY_PREFIX + recordDTO.getBizId();
        //3.执行srem命令
        Long result = redisTemplate.opsForSet().remove(key, userId.toString());
        return result != null && result > 0;

    }

    /**
     * 执行点赞
     * @param recordDTO 点赞业务类型
     * @return 是否取消点赞
     */
    private Boolean like(LikeRecordFormDTO recordDTO) {
        // 1.获取用户id
        Long userId = UserContext.getUser();
        // 2.获取Key
        String key = LIKE_BIZ_KEY_PREFIX + recordDTO.getBizId();
        // 3.执行SADD命令
        Long result = redisTemplate.opsForSet().add(key, userId.toString());
        return result != null && result > 0;
    }
}
